package com.ldf.media.callback;

import com.aizuda.zlm4j.callback.IMKSourceFindCallBack;
import com.aizuda.zlm4j.callback.IMKStreamChangeCallBack;
import com.aizuda.zlm4j.structure.MK_MEDIA_SOURCE;
import com.ldf.media.api.service.IApiService;
import com.ldf.media.api.service.ITranscodeService;
import com.ldf.media.api.service.impl.ApiServiceImpl;
import com.ldf.media.config.MediaServerConfig;
import com.ldf.media.context.MediaServerContext;
import com.sun.jna.CallbackThreadInitializer;
import com.sun.jna.Native;
import com.sun.jna.Pointer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static com.ldf.media.constants.MediaServerConstants.*;

/**
 * 注册或反注册MediaSource事件广播
 *
 * @author lidaofu
 * @since 2023/11/23
 **/
@Component
@Slf4j
public class MKStreamChangeCallBack implements IMKStreamChangeCallBack, ApplicationContextAware {
    @Autowired
    private ITranscodeService transcodeService;
    private ApplicationContext applicationContext;
    @Autowired
    private MediaServerConfig config;
    
    // 存储流的最后活跃时间
    public static final Map<String, Long> streamLastActiveTime = new ConcurrentHashMap<>();
    
    // 存储流的播放人数
    public static final Map<String, Integer> playerCountMap = new ConcurrentHashMap<>();

    public MKStreamChangeCallBack() {
        Native.setCallbackThreadInitializer(this, new CallbackThreadInitializer(true, false, "MediaStreamChangeThread"));
    }

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    /**
     * 注册或反注册MediaSource事件广播
     *
     * @param regist 注册为1，注销为0
     * @param sender 该MediaSource对象
     */
    public void invoke(int regist, MK_MEDIA_SOURCE sender) {
        //这里进行流状态处理
        String stream = MediaServerContext.ZLM_API.mk_media_source_get_stream(sender);
        String app = MediaServerContext.ZLM_API.mk_media_source_get_app(sender);
        String schema = MediaServerContext.ZLM_API.mk_media_source_get_schema(sender);
        //如果是regist是注销情况下无法获取流详细信息如观看人数等
        log.info("【MediaServer】APP:{} 流:{} 协议：{} {}", app, stream, schema, regist == 1 ? "注册" : "注销");
        if (schema.equals("rtmp")&&regist==0){
            transcodeService.stopTranscode(stream);
        }
    }

    /**
     * 初始化定时任务
     */
    @PostConstruct
    public void init() {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);

        scheduler.scheduleAtFixedRate(() -> {
            try {
                // 清理无效流
                cleanupInvalidStreams();
                
                // 清理拉流代理
                cleanupProxyStreams();

                // 打印当前播放流数量
                log.info("当前正在播放的流数量：{}", getActiveStreamCount());
            } catch (Exception e) {
                log.error("流状态监控任务发生异常", e);
            }
        }, 0, 30, TimeUnit.SECONDS); // 每30秒检查一次
        
        // 每分钟检查一次录像状态
        scheduler.scheduleAtFixedRate(() -> {
            try {
                log.info("检查录像状态");
            } catch (Exception e) {
                log.error("录像状态检查任务发生异常", e);
            }
        }, 0, 1, TimeUnit.MINUTES);
    }

    /**
     * 构建唯一的流标识符
     */
    private String buildStreamId(MK_MEDIA_SOURCE source) {
        return String.format("%s/%s",
                MediaServerContext.ZLM_API.mk_media_source_get_app(source),
                MediaServerContext.ZLM_API.mk_media_source_get_stream(source));
    }

    /**
     * 获取当前活跃的流数量
     */
    public static int getActiveStreamCount() {
        // 使用ZLMediaKit API直接获取活跃流数量，而不是依赖streamLastActiveTime
        AtomicInteger count = new AtomicInteger(0);
        MediaServerContext.ZLM_API.mk_media_source_for_each(
            null, 
            new IMKSourceFindCallBack() {
                @Override
                public void invoke(Pointer user_data, MK_MEDIA_SOURCE ctx) {
                    try {
                        // 获取流的基本信息
                        String schema = MediaServerContext.ZLM_API.mk_media_source_get_schema(ctx);
                        String app = MediaServerContext.ZLM_API.mk_media_source_get_app(ctx);
                        String stream = MediaServerContext.ZLM_API.mk_media_source_get_stream(ctx);
                        
                        // 排除系统内部流（如__defaultVhost__）
                        if ("__defaultVhost__".equals(app)) {
                            return;
                        }
                        
                        // 检查流是否活跃（有观看者或者存活时间较短）
                        int totalReaderCount = MediaServerContext.ZLM_API.mk_media_source_get_total_reader_count(ctx);
                        long aliveSecond = MediaServerContext.ZLM_API.mk_media_source_get_alive_second(ctx);
                        
                        // 如果有观看者或者存活时间小于10分钟，则认为是活跃流
                        if (totalReaderCount > 0 || aliveSecond < 600) {
                            count.incrementAndGet();
                        }
                    } catch (Exception e) {
                        log.error("统计活跃流时发生异常", e);
                    }
                }
            }, 
            null, null, null, null
        );
        return count.get();
    }

    /**
     * 清理超时流
     */
    private void cleanupInvalidStreams() {
        streamLastActiveTime.forEach((streamId, lastActiveTime) -> {
            if (isStreamTimeout(lastActiveTime)) {
                log.info("cleanupInvalidStreams:流 {} 已超时", streamId);
                // 查找流
                MK_MEDIA_SOURCE source = MediaServerContext.ZLM_API.mk_media_source_find2(
                        "rtmp", DEFAULT_VHOST,
                        streamId.split("/")[0],
                        streamId.split("/")[1], 0);

                if (source != null) {
                    //查看流是否有人观看：
                    int totalReaderCount = MediaServerContext.ZLM_API.mk_media_source_get_total_reader_count(source);
                    long aliveSecond = MediaServerContext.ZLM_API.mk_media_source_get_alive_second(source);
                    
                    // 判断条件：无人观看且存活时间超过5分钟
                    if (totalReaderCount <= 0 && aliveSecond > 300) {
                        // 检查是否是拉流代理创建的流
                        int originType = MediaServerContext.ZLM_API.mk_media_source_get_origin_type(source);
                        if (originType == 4) { // 拉流代理的originType为4
                            String app = MediaServerContext.ZLM_API.mk_media_source_get_app(source);
                            String stream = MediaServerContext.ZLM_API.mk_media_source_get_stream(source);
                            log.info("检测到超时且无人观看的拉流代理流，准备关闭: {}/{}, 观看人数: {}, 存活时间: {}秒", 
                                    app, stream, totalReaderCount, aliveSecond);
                            
                            // 获取拉流代理key并关闭
                            IApiService apiService = applicationContext.getBean(IApiService.class);
                            String proxyKey = apiService.getProxyKeyByStream(app, stream);
                            if (proxyKey != null) {
                                log.info("关闭超时且无人观看的拉流代理流: {}/{}, proxyKey: {}", app, stream, proxyKey);
                                apiService.delStreamProxy(proxyKey);
                            } else {
                                log.warn("未找到超时拉流代理流的proxyKey: {}/{}", app, stream);
                            }
                        } else {
                            log.info("非拉流代理流，不自动关闭: {}", streamId);
                        }
                    } else {
                        log.info("流 {} 仍有观众({}人)或存活时间不足5分钟({}秒)，不关闭", streamId, totalReaderCount, aliveSecond);
                    }

                    // 更新计数器
                    playerCountMap.remove(streamId);

                    log.info("流 {} 已超时并关闭", streamId);
                }
            }
        });
    }
    
    /**
     * 清理拉流代理流（新方法）
     * 直接通过ZLMediaKit API遍历所有流，检查是否为拉流代理且满足关闭条件
     */
    private void cleanupProxyStreams() {
        log.debug("开始执行拉流代理清理任务");
        try {
            MediaServerContext.ZLM_API.mk_media_source_for_each(
                null,
                new IMKSourceFindCallBack() {
                    @Override
                    public void invoke(Pointer user_data, MK_MEDIA_SOURCE source) {
                        try {
                            // 检查是否是拉流代理创建的流 (originType == 4)
                            int originType = MediaServerContext.ZLM_API.mk_media_source_get_origin_type(source);
                            if (originType == 4) {
                                String app = MediaServerContext.ZLM_API.mk_media_source_get_app(source);
                                String stream = MediaServerContext.ZLM_API.mk_media_source_get_stream(source);
                                String schema = MediaServerContext.ZLM_API.mk_media_source_get_schema(source);
                                
                                // 获取观看人数和存活时间
                                int totalReaderCount = MediaServerContext.ZLM_API.mk_media_source_get_total_reader_count(source);
                                long aliveSecond = MediaServerContext.ZLM_API.mk_media_source_get_alive_second(source);
                                
                                log.debug("检查拉流代理流: {}/{}/{}, 观看人数: {}, 存活时间: {}秒", 
                                         schema, app, stream, totalReaderCount, aliveSecond);
                                
                                // 判断是否满足关闭条件：无人观看且存活时间超过5分钟
                                if (totalReaderCount <= 0 && aliveSecond > 300) {
                                    log.info("检测到超时且无人观看的拉流代理流，准备关闭: {}/{}/{}, 观看人数: {}, 存活时间: {}秒", 
                                            schema, app, stream, totalReaderCount, aliveSecond);
                                    
                                    // 获取拉流代理key并关闭
                                    IApiService apiService = applicationContext.getBean(IApiService.class);
                                    String proxyKey = apiService.getProxyKeyByStream(app, stream);
                                    if (proxyKey != null) {
                                        log.info("关闭超时且无人观看的拉流代理流: {}/{}/{}, proxyKey: {}", 
                                                schema, app, stream, proxyKey);
                                        apiService.delStreamProxy(proxyKey);
                                    } else {
                                        log.warn("未找到超时拉流代理流的proxyKey: {}/{}/{}", schema, app, stream);
                                    }
                                } else {
                                    log.debug("拉流代理流 {}/{}/{} 不满足关闭条件: 观看人数({})或存活时间({}秒)", 
                                             schema, app, stream, totalReaderCount, aliveSecond);
                                }
                            }
                        } catch (Exception e) {
                            log.error("处理单个流时发生异常", e);
                        }
                    }
                },
                null, null, null, null
            );
            log.debug("拉流代理清理任务执行完成");
        } catch (Exception e) {
            log.error("执行拉流代理清理任务时发生异常", e);
        }
    }

    /**
     * 判断流是否超时 (10分钟)
     */
    private static boolean isStreamTimeout(long lastActiveTime) {
        return System.currentTimeMillis() - lastActiveTime > 600000; // 10分钟超时
    }
}